package org.apache.rocketmq.store;



import com.google.common.collect.Lists;
import org.apache.rocketmq.common.BoundaryType;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.store.logfile.DefaultMappedFile;
import org.apache.rocketmq.store.logfile.MappedFile;

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.stream.Stream;


/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/29
 * @方法描述：内存映射文件管理器
 */
public class MappedFileQueue implements Swappable {

    private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);

    private static final Logger LOG_ERROR = LoggerFactory.getLogger(LoggerName.STORE_ERROR_LOGGER_NAME);

    //CommitLog或ConsumeQueue文件的存储目录
    protected final String storePath;

    //单个内存映射文件的大小
    protected final int mappedFileSize;

    //存放所有内存映射文件都西昂的集合
    protected final CopyOnWriteArrayList<MappedFile> mappedFiles = new CopyOnWriteArrayList<>();

    //内存映射文件分配器
    protected final AllocateMappedFileService allocateMappedFileService;

    //消息最后刷盘位置
    protected long flushedWhere = 0;

    //消息最新提交到操作系统缓存的位置
    protected long committedWhere = 0;

    //消息最后的存储时间
    protected volatile long storeTimestamp = 0;

    //构造方法
    public MappedFileQueue(final String storePath, int mappedFileSize,
                           AllocateMappedFileService allocateMappedFileService) {
        this.storePath = storePath;
        this.mappedFileSize = mappedFileSize;
        this.allocateMappedFileService = allocateMappedFileService;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：内存映射文件队列启动时的自检方法，就是看看各个内存映射文件是否是连续的
     */
    public void checkSelf() {
        //创建副本数据
        List<MappedFile> mappedFiles = new ArrayList<>(this.mappedFiles);
        if (!mappedFiles.isEmpty()) {
            Iterator<MappedFile> iterator = mappedFiles.iterator();
            MappedFile pre = null;
            //遍历所有内存映射文件
            while (iterator.hasNext()) {
                MappedFile cur = iterator.next();
                //下面就是当前文件和上一个文件偏移量做对比的操作
                if (pre != null) {
                    //计算当前文件与上一个文件的偏移量差，如果偏移量差不是单个内存映射文件的大小
                    //则意味着文件可能被损坏了
                    if (cur.getFileFromOffset() - pre.getFileFromOffset() != this.mappedFileSize){
                        //记录错误日志
                        LOG_ERROR.error("[BUG]The mappedFile queue's data is damaged, the adjacent mappedFile's offset don't match. pre file {}, cur file {}",
                                pre.getFileName(), cur.getFileName());
                    }
                }//把当前文件赋值给上一个文件，进入下一次循环，使用上一个文件和即将得到的当前文件做对比
                pre = cur;
            }
        }
    }


    //在这里我跟大家解释一下，该类中有很多方法在第十四版本代码中没有用到，所以我就不添加注释了，要不然要写的注释就太多了，等后面用到这些方法时，我再添加注释
    //请大家谅解一下

    public MappedFile getConsumeQueueMappedFileByTime(final long timestamp, CommitLog commitLog, BoundaryType boundaryType) {
        Object[] mfs = copyMappedFiles(0);
        if (null == mfs) {
            return null;
        }
        for (int i = mfs.length - 1; i >= 0; i--) {
            DefaultMappedFile mappedFile = (DefaultMappedFile) mfs[i];
            if (mappedFile.getStartTimestamp() < 0) {
                SelectMappedBufferResult selectMappedBufferResult = mappedFile.selectMappedBuffer(0, ConsumeQueue.CQ_STORE_UNIT_SIZE);
                if (null != selectMappedBufferResult) {
                    try {
                        ByteBuffer buffer = selectMappedBufferResult.getByteBuffer();
                        long physicalOffset = buffer.getLong();
                        int messageSize = buffer.getInt();
                        long messageStoreTime = commitLog.pickupStoreTimestamp(physicalOffset, messageSize);
                        if (messageStoreTime > 0) {
                            mappedFile.setStartTimestamp(messageStoreTime);
                        }
                    } finally {
                        selectMappedBufferResult.release();
                    }
                }
            }
            if (i < mfs.length - 1 && mappedFile.getStopTimestamp() < 0) {
                SelectMappedBufferResult selectMappedBufferResult = mappedFile.selectMappedBuffer(mappedFileSize - ConsumeQueue.CQ_STORE_UNIT_SIZE, ConsumeQueue.CQ_STORE_UNIT_SIZE);
                if (null != selectMappedBufferResult) {
                    try {
                        ByteBuffer buffer = selectMappedBufferResult.getByteBuffer();
                        long physicalOffset = buffer.getLong();
                        int messageSize = buffer.getInt();
                        long messageStoreTime = commitLog.pickupStoreTimestamp(physicalOffset, messageSize);

                        if (messageStoreTime > 0) {
                            mappedFile.setStopTimestamp(messageStoreTime);
                        }
                    } finally {
                        selectMappedBufferResult.release();
                    }
                }
            }
        }

        switch (boundaryType) {
            case LOWER: {
                for (int i = 0; i < mfs.length; i++) {
                    DefaultMappedFile mappedFile = (DefaultMappedFile) mfs[i];
                    if (i < mfs.length - 1) {
                        long stopTimestamp = mappedFile.getStopTimestamp();
                        if (stopTimestamp >= timestamp) {
                            return mappedFile;
                        }
                    }
                    if (i == mfs.length - 1) {
                        return mappedFile;
                    }
                }
            }
            case UPPER: {
                for (int i = mfs.length - 1; i >= 0; i--) {
                    DefaultMappedFile mappedFile = (DefaultMappedFile) mfs[i];
                    if (mappedFile.getStartTimestamp() <= timestamp) {
                        return mappedFile;
                    }
                }
            }
            default: {
                log.warn("Unknown boundary type");
                break;
            }
        }
        return null;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：根据文件最后修改时间找到指定内存映射文件的方法
     */
    public MappedFile getMappedFileByTime(final long timestamp) {
        //得到当前内存映射文件的拷贝数据，方法参数0表示内存映射队列中的内存映射文件数量不能小于等于0
        //也就是说内存映射文件队列中必须有数据才能执行拷贝操作
        Object[] mfs = this.copyMappedFiles(0);
        if (null == mfs)
            return null;
        //遍历啊内存映射文件数据
        for (int i = 0; i < mfs.length; i++) {
            MappedFile mappedFile = (MappedFile) mfs[i];
            //判断遍历到的内存映射文件的最后修改时间是否大于目标时间戳
            if (mappedFile.getLastModifiedTimestamp() >= timestamp) {
                //大于目标时间则返回遍历到的内存映射文件
                return mappedFile;
            }
        }
        //如果没找到目标文件，则返回最新的内存映射文件
        return (MappedFile) mfs[mfs.length - 1];
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到当前内存映射文件队列的拷贝数据
     */
    protected Object[] copyMappedFiles(final int reservedMappedFiles) {
        Object[] mfs;
        if (this.mappedFiles.size() <= reservedMappedFiles) {
            return null;
        }
        mfs = this.mappedFiles.toArray();
        return mfs;
    }


    public void truncateDirtyFiles(long offset) {
        List<MappedFile> willRemoveFiles = new ArrayList<>();
        for (MappedFile file : this.mappedFiles) {
            long fileTailOffset = file.getFileFromOffset() + this.mappedFileSize;
            if (fileTailOffset > offset) {
                if (offset >= file.getFileFromOffset()) {
                    file.setWrotePosition((int) (offset % this.mappedFileSize));
                    file.setCommittedPosition((int) (offset % this.mappedFileSize));
                    file.setFlushedPosition((int) (offset % this.mappedFileSize));
                }
                else {
                    file.destroy(1000);
                    willRemoveFiles.add(file);
                }
            }
        }
        this.deleteExpiredFile(willRemoveFiles);
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：从内存映射文件集合中删除指定内存映射文件的方法
     */
    void deleteExpiredFile(List<MappedFile> files) {
        if (!files.isEmpty()) {
            Iterator<MappedFile> iterator = files.iterator();
            while (iterator.hasNext()) {
                MappedFile cur = iterator.next();
                if (!this.mappedFiles.contains(cur)) {
                    iterator.remove();
                    log.info("This mappedFile {} is not contained by mappedFiles, so skip it.", cur.getFileName());
                }
            }
            try {
                if (!this.mappedFiles.removeAll(files)) {
                    log.error("deleteExpiredFile remove failed.");
                }
            } catch (Exception e) {
                log.error("deleteExpiredFile has exception.", e);
            }
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：加载存储路径下的本地文件到内存映射队列中的方法
     */
    public boolean load() {
        //创建存储目录文件对象，文件路径就是：/store/commitlog
        File dir = new File(this.storePath);
        //获取文件夹下的所有文件
        File[] ls = dir.listFiles();
        if (ls != null) {
            //那文件加载到内存映射文件中
            return doLoad(Arrays.asList(ls));
        }
        return true;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：加载本地文件到内存映射队列的方法
     */
    public boolean doLoad(List<File> files) {
        //根据内存映射文件名称从小到大排序
        files.sort(Comparator.comparing(File::getName));
        //遍历所有内存映射文件
        for (int i = 0; i < files.size(); i++) {
            //得到文件本身
            File file = files.get(i);
            //如果当前文件是文件夹则跳过本次循环
            if (file.isDirectory()) {
                continue;
            }
            //这里就是判断当前遍历到的文件是不是文件队列的最后一个文件
            if (file.length() == 0 && i == files.size() - 1) {
                //如果是队列最后一个文件，并且是空文件，那就删除这个文件
                boolean ok = file.delete();
                //记录告警日志
                log.warn("{} size is 0, auto delete. is_ok: {}", file, ok);
                continue;
            }
            //校验文件大小是否等于用户配置的大小
            if (file.length() != this.mappedFileSize) {
                //不符合则记录告警日志
                log.warn(file + "\t" + file.length()
                        + " length not matched message store config value, please check it manually");
                //并且返回false，终止加载内存映射文件操作
                return false;
            }
            try {
                //在这里创建内存映射文件对象，创建对象的过程会执行初始化操作，完成内存映射操作
                MappedFile mappedFile = new DefaultMappedFile(file.getPath(), mappedFileSize);
                //设置内存映射文件的写入位置
                mappedFile.setWrotePosition(this.mappedFileSize);
                //设置刷盘位置
                mappedFile.setFlushedPosition(this.mappedFileSize);
                //设置消息已提交位置
                mappedFile.setCommittedPosition(this.mappedFileSize);
                //把文件加入内存映射队列，这里大家可能会困惑，为什么从本地加载的内存映射文件的消息写入位置，消息刷盘位置，消息提交位置
                //都设置成最满呢？如果本地最后一个文件没有写满数据，但是设置成最满了，那后续还怎么向该文件写入新数据呢？这不是浪费吗？
                //我知道大家肯定会有这个困惑，别急朋友们，请大家先记住这个问题，等后面两个版本代码，我们实现了消息存储引擎初始化加载功能
                //大家就知道究竟是怎么回事了，到时候功能就更加完善，大家的困惑也就消除了
                this.mappedFiles.add(mappedFile);
                log.info("load " + file.getPath() + " OK");
            } catch (IOException e) {
                log.error("load file " + file + " error", e);
                return false;
            }
        }
        return true;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：计算未刷盘的消息的方法
     */
    public long howMuchFallBehind() {
        if (this.mappedFiles.isEmpty())
            return 0;

        //得到已刷盘的消息的偏移量
        long committed = this.getFlushedWhere();
        if (committed != 0) {
            //得到内存映射队列的最后一个文件
            MappedFile mappedFile = this.getLastMappedFile(0, false);
            if (mappedFile != null) {
                //最后一个文件写入消息的偏移量如果大于已刷盘的消息的偏移量，就意味着还有数据没有刷盘
                //返回这个数据即可
                return (mappedFile.getFileFromOffset() + mappedFile.getWrotePosition()) - committed;
            }
        }
        return 0;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：的到最后一个内存映射文件的方法
     */
    public MappedFile getLastMappedFile(final long startOffset, boolean needCreate) {
        long createOffset = -1;
        //得到文件队列的最后一个文件
        MappedFile mappedFileLast = getLastMappedFile();
        if (mappedFileLast == null) {
            //如果最后一个文件不存在，意味着队列中还没有数据，这时候就要创建新的文件了
            //在这里计算要创建的内存映射文件的起始偏移量
            //和操作系统缓存页对齐
            createOffset = startOffset - (startOffset % this.mappedFileSize);
        }
        //如果最后一个文件存在，但是已经写满了，这时候也要创建新的文件
        if (mappedFileLast != null && mappedFileLast.isFull()) {
            //得到要创建的新文件的起始偏移量，就是最后一个文件起始偏移量加上文件大小
            createOffset = mappedFileLast.getFileFromOffset() + this.mappedFileSize;
        }
        //判断文件偏移量是否有效，以及是否需要创建新文件
        if (createOffset != -1 && needCreate) {
            //需要创建则在这里执行创建新文件的操作
            return tryCreateMappedFile(createOffset);
        }
        return mappedFileLast;
    }

    //判断内存映射队列是否为空的方法
    public boolean isMappedFilesEmpty() {
        return this.mappedFiles.isEmpty();
    }

    //判断mappedFiles集合是否为空或者队列中最后一个文件是否已满的方法
    public boolean isEmptyOrCurrentFileFull() {
        MappedFile mappedFileLast = getLastMappedFile();
        if (mappedFileLast == null) {
            return true;
        }
        if (mappedFileLast.isFull()) {
            return true;
        }
        return false;
    }

    //判断是否需要创建新文件的方法，方法参数是要写入内存映射文件的消息大小
    public boolean shouldRoll(final int msgSize) {
        if (isEmptyOrCurrentFileFull()) {
            return true;
        }
        //得到队列中最后一个文件
        MappedFile mappedFileLast = getLastMappedFile();
        //当最新的文件可写空间不足了，就要创建新文件
        if (mappedFileLast.getWrotePosition() + msgSize > mappedFileLast.getFileSize()) {
            return true;
        }

        return false;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：尝试创建新的内存映射文件的方法
     */
    public MappedFile tryCreateMappedFile(long createOffset) {
        //得到要创建的文件路径，createOffset就是文件起始偏移量，也就是文件的名称
        //文件路径可以是这样：/store/commitlog/00000000000000000000
        String nextFilePath = this.storePath + File.separator + UtilAll.offset2FileName(createOffset);
        //得到下下个文件的路径
        String nextNextFilePath = this.storePath + File.separator + UtilAll.offset2FileName(createOffset + this.mappedFileSize);
        //执行创建内存映射文件的操作，这里可以看到，每次创建内存映射文件的时候，会一次创建两个
        return doCreateMappedFile(nextFilePath, nextNextFilePath);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：真正创建内存映射文件的方法
     */
    protected MappedFile doCreateMappedFile(String nextFilePath, String nextNextFilePath) {
        MappedFile mappedFile = null;
        //判断是否使用了内存映射文件分配器
        if (this.allocateMappedFileService != null) {
            //如果使用了则提交创建内存映射文件请求到分配器中
            //让分配器创建内存映射文件
            mappedFile = this.allocateMappedFileService.putRequestAndReturnMappedFile(nextFilePath, nextNextFilePath, this.mappedFileSize);
        } else {
            try {//如果没有使用内存映射文件分配器，则直接在这里创建内存映射文件
                mappedFile = new DefaultMappedFile(nextFilePath, this.mappedFileSize);
            } catch (IOException e) {
                log.error("create mappedFile exception", e);
            }
        }
        if (mappedFile != null) {
            if (this.mappedFiles.isEmpty()) {
                //如果这是队列中第一个创建的内存映射文件，那就给这个文件设置首个创建标志
                mappedFile.setFirstCreateInQueue(true);
            }
            //创建成功后把内存映射文件添加到队列中
            this.mappedFiles.add(mappedFile);
        }

        return mappedFile;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到队列中最新的内存映射文件的方法
     */
    public MappedFile getLastMappedFile(final long startOffset) {
        return getLastMappedFile(startOffset, true);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到对立中最后一个内存映射文件的方法
     */
    public MappedFile getLastMappedFile() {
        MappedFile mappedFileLast = null;
        while (!this.mappedFiles.isEmpty()) {
            try {
                //得到文件队列的最后一个文件
                mappedFileLast = this.mappedFiles.get(this.mappedFiles.size() - 1);
                break;
            } catch (IndexOutOfBoundsException e) {

            } catch (Exception e) {
                log.error("getLastMappedFile has exception.", e);
                break;
            }
        }
        return mappedFileLast;
    }


    public boolean resetOffset(long offset) {
        MappedFile mappedFileLast = getLastMappedFile();
        if (mappedFileLast != null) {
            long lastOffset = mappedFileLast.getFileFromOffset() + mappedFileLast.getWrotePosition();
            long diff = lastOffset - offset;
            final int maxDiff = this.mappedFileSize * 2;
            if (diff > maxDiff)
                return false;
        }
        ListIterator<MappedFile> iterator = this.mappedFiles.listIterator(mappedFiles.size());
        List<MappedFile> toRemoves = new ArrayList<>();

        while (iterator.hasPrevious()) {
            mappedFileLast = iterator.previous();
            if (offset >= mappedFileLast.getFileFromOffset()) {
                int where = (int) (offset % mappedFileLast.getFileSize());
                mappedFileLast.setFlushedPosition(where);
                mappedFileLast.setWrotePosition(where);
                mappedFileLast.setCommittedPosition(where);
                break;
            } else {
                toRemoves.add(mappedFileLast);
            }
        }
        if (!toRemoves.isEmpty()) {
            this.mappedFiles.removeAll(toRemoves);
        }

        return true;
    }


    public long getMinOffset() {
        if (!this.mappedFiles.isEmpty()) {
            try {
                return this.mappedFiles.get(0).getFileFromOffset();
            } catch (IndexOutOfBoundsException e) {

            } catch (Exception e) {
                log.error("getMinOffset has exception.", e);
            }
        }
        return -1;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到当前内存映射文件中消息最大偏移量
     */
    public long getMaxOffset() {
        //得到最后的内存映射文件
        MappedFile mappedFile = getLastMappedFile();
        if (mappedFile != null) {
            //得到最新内存映射文件的可读消息最大偏移量
            return mappedFile.getFileFromOffset() + mappedFile.getReadPosition();
        }
        return 0;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到当前内存映射文件中写入的消息最大偏移量
     */
    public long getMaxWrotePosition() {
        MappedFile mappedFile = getLastMappedFile();
        if (mappedFile != null) {
            return mappedFile.getFileFromOffset() + mappedFile.getWrotePosition();
        }
        return 0;
    }


    //计算待提交到操作系统缓存页的消息大小的方法
    public long remainHowManyDataToCommit() {
        return getMaxWrotePosition() - getCommittedWhere();
    }


    //计算待刷盘的消息大小的方法
    public long remainHowManyDataToFlush() {
        return getMaxOffset() - this.getFlushedWhere();
    }


    //删除最后一个内存映射文件的方法
    public void deleteLastMappedFile() {
        MappedFile lastMappedFile = getLastMappedFile();
        if (lastMappedFile != null) {
            lastMappedFile.destroy(1000);
            this.mappedFiles.remove(lastMappedFile);
            log.info("on recover, destroy a logic mapped file " + lastMappedFile.getFileName());
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：删除过期内存映射文件的方法，该方法在第十四版本代码中先不添加注释
     */
    public int deleteExpiredFileByTime(final long expiredTime,
                                       final int deleteFilesInterval,
                                       final long intervalForcibly,
                                       final boolean cleanImmediately,
                                       final int deleteFileBatchMax) {
        Object[] mfs = this.copyMappedFiles(0);
        if (null == mfs)
            return 0;
        int mfsLength = mfs.length - 1;
        int deleteCount = 0;
        List<MappedFile> files = new ArrayList<>();
        int skipFileNum = 0;
        if (null != mfs) {
            checkSelf();
            for (int i = 0; i < mfsLength; i++) {
                MappedFile mappedFile = (MappedFile) mfs[i];
                long liveMaxTimestamp = mappedFile.getLastModifiedTimestamp() + expiredTime;
                if (System.currentTimeMillis() >= liveMaxTimestamp || cleanImmediately) {
                    if (skipFileNum > 0) {
                        log.info("Delete CommitLog {} but skip {} files", mappedFile.getFileName(), skipFileNum);
                        skipFileNum = 0;
                    }
                    if (mappedFile.destroy(intervalForcibly)) {
                        files.add(mappedFile);
                        deleteCount++;
                        if (files.size() >= deleteFileBatchMax) {
                            break;
                        }
                        if (deleteFilesInterval > 0 && (i + 1) < mfsLength) {
                            try {
                                Thread.sleep(deleteFilesInterval);
                            } catch (InterruptedException e) {

                            }
                        }
                    } else {
                        break;
                    }
                } else {
                    skipFileNum++;
                    break;
                }
            }
        }
        deleteExpiredFile(files);
        return deleteCount;
    }



    public int deleteExpiredFileByOffset(long offset, int unitSize) {
        Object[] mfs = this.copyMappedFiles(0);
        List<MappedFile> files = new ArrayList<>();
        int deleteCount = 0;
        if (null != mfs) {
            int mfsLength = mfs.length - 1;
            for (int i = 0; i < mfsLength; i++) {
                boolean destroy = false;
                MappedFile mappedFile = (MappedFile) mfs[i];
                SelectMappedBufferResult result = mappedFile.selectMappedBuffer(this.mappedFileSize - unitSize);
                if (result != null) {
                    long maxOffsetInLogicQueue = result.getByteBuffer().getLong();
                    result.release();
                    destroy = maxOffsetInLogicQueue < offset;
                    if (destroy) {
                        log.info("physic min offset " + offset + ", logics in current mappedFile max offset "
                                + maxOffsetInLogicQueue + ", delete it");
                    }
                }
                else if (!mappedFile.isAvailable()) {
                    log.warn("Found a hanged consume queue file, attempting to delete it.");
                    destroy = true;
                }
                else {
                    log.warn("this being not executed forever.");
                    break;
                }
                if (destroy && mappedFile.destroy(1000 * 60)) {
                    files.add(mappedFile);
                    deleteCount++;
                } else {
                    break;
                }
            }
        }
        deleteExpiredFile(files);
        return deleteCount;
    }


    public int deleteExpiredFileByOffsetForTimerLog(long offset, int checkOffset, int unitSize) {
        Object[] mfs = this.copyMappedFiles(0);
        List<MappedFile> files = new ArrayList<>();
        int deleteCount = 0;
        if (null != mfs) {
            int mfsLength = mfs.length - 1;
            for (int i = 0; i < mfsLength; i++) {
                boolean destroy = false;
                MappedFile mappedFile = (MappedFile) mfs[i];
                SelectMappedBufferResult result = mappedFile.selectMappedBuffer(checkOffset);
                try {
                    if (result != null) {
                        int position = result.getByteBuffer().position();
                        int size = result.getByteBuffer().getInt();
                        result.getByteBuffer().getLong();
                        int magic = result.getByteBuffer().getInt();
                        if (size == unitSize && (magic | 0xF) == 0xF) {
                            result.getByteBuffer().position(position + MixAll.UNIT_PRE_SIZE_FOR_MSG);
                            long maxOffsetPy = result.getByteBuffer().getLong();
                            destroy = maxOffsetPy < offset;
                            if (destroy) {
                                log.info("physic min commitlog offset " + offset + ", current mappedFile's max offset "
                                        + maxOffsetPy + ", delete it");
                            }
                        } else {
                            log.warn("Found error data in [{}] checkOffset:{} unitSize:{}", mappedFile.getFileName(),
                                    checkOffset, unitSize);
                        }
                    } else if (!mappedFile.isAvailable()) {
                        log.warn("Found a hanged consume queue file, attempting to delete it.");
                        destroy = true;
                    } else {
                        log.warn("this being not executed forever.");
                        break;
                    }
                } finally {
                    if (null != result) {
                        result.release();
                    }
                }
                if (destroy && mappedFile.destroy(1000 * 60)) {
                    files.add(mappedFile);
                    deleteCount++;
                } else {
                    break;
                }
            }
        }
        deleteExpiredFile(files);
        return deleteCount;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：刷新消息到本地硬盘的方法
     */
    public boolean flush(final int flushLeastPages) {
        boolean result = true;
        //根据当前已刷盘位置查找对应的要进行数据刷盘的内存映射文件
        MappedFile mappedFile = this.findMappedFileByOffset(
                //当前已刷盘的消息偏移量
                this.getFlushedWhere(),
                //是否刷盘偏移量为0，如果为0说明还没有消息刷盘，那就要从文件队列的第一个文件开始刷盘
                this.getFlushedWhere() == 0
        );
        if (mappedFile != null) {
            //得到最新消息的存储时间
            long tmpTimeStamp = mappedFile.getStoreTimestamp();
            //执行消息刷盘操作，返回的是消息刷盘之后最新的刷盘偏移量
            int offset = mappedFile.flush(flushLeastPages);
            //得到整个CommitLog文件的全局刷盘位置
            long where = mappedFile.getFileFromOffset() + offset;
            //这里判断的其实是是否刷盘完毕
            //判断新的刷盘全局位置和旧的刷盘全局位置是否一样，如果一样意味着本次操作没有数据刷新到硬盘
            //可能意味着上次刷盘操作已经是最新的了
            //如果不一行则意味着确实有数据刷盘了
            result = where == this.getFlushedWhere();
            //更新全局刷盘位置
            this.setFlushedWhere(where);
            if (0 == flushLeastPages) {
                this.setStoreTimestamp(tmpTimeStamp);
            }
        }//返回ture意味着数据已经刷盘完毕了，返回false意味着可能还有数据待刷新
        return result;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：提交数据到操作系统缓存页的方法，这个方法和上面方法的逻辑一样，只不过就是刷新变成了提交，所以我就不添加注释了，大家类比着看看就行
     */
    public synchronized boolean commit(final int commitLeastPages) {
        boolean result = true;
        MappedFile mappedFile = this.findMappedFileByOffset(
                this.getCommittedWhere(),
                this.getCommittedWhere() == 0
        );
        if (mappedFile != null) {
            int offset = mappedFile.commit(commitLeastPages);
            long where = mappedFile.getFileFromOffset() + offset;
            result = where == this.getCommittedWhere();
            this.setCommittedWhere(where);
        }
        return result;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：根据当前已刷盘位置查找对应的要进行数据刷盘的内存映射文件的方法，第二个方法参数returnFirstOnNotFound表示在没找到映射文件的情况下是否返回文件队列第一个文件的标志
     */
    public MappedFile findMappedFileByOffset(final long offset, final boolean returnFirstOnNotFound) {
        try {
            //得到文件队列中的首个文件
            MappedFile firstMappedFile = this.getFirstMappedFile();
            //得文件队列中最后一个文件
            MappedFile lastMappedFile = this.getLastMappedFile();
            //当首末映射文件都存在时进入以下分支
            if (firstMappedFile != null && lastMappedFile != null) {
                //计算首文件的起始偏移量
                final long firstFileStartOffset = firstMappedFile.getFileFromOffset();
                //计算最后一个文件最后可写入消息的偏移量，这里得到的就是整个CommitLog文件消息偏移量的合法区间
                final long lastFileEndOffset = lastMappedFile.getFileFromOffset() + this.mappedFileSize;
                //判断要定位的消息偏移量是否不在合法范围之内
                if (offset < firstFileStartOffset || offset >= lastFileEndOffset) {
                    //不在则记录告警日志
                    LOG_ERROR.warn("Offset not matched. Request offset: {}, firstOffset: {}, lastOffset: {}, mappedFileSize: {}, mappedFiles count: {}",
                            offset,
                            firstFileStartOffset,
                            lastFileEndOffset,
                            this.mappedFileSize,
                            this.mappedFiles.size());
                } else {
                    //执行到这里意味着要定位的消息偏移量在合法范围内，那就根据消息偏移量计算对应文件的索引，找到这个消息偏移量在哪个文件中
                    //计算公式：index = (目标偏移/文件尺寸) - (首文件起始偏移/文件尺寸)
                    int index = (int) ((offset / this.mappedFileSize) - (firstMappedFile.getFileFromOffset() / this.mappedFileSize));
                    MappedFile targetFile = null;
                    try {
                        //根据所以得到具体的内存映射文件
                        targetFile = this.mappedFiles.get(index);
                    } catch (Exception ignored) {

                    }
                    //判断找到的这个文件是否是合法文件，就是校验一下目标文件的消息偏移量是否合法
                    if (targetFile != null
                            && offset >= targetFile.getFileFromOffset()
                            && offset < targetFile.getFileFromOffset() + this.mappedFileSize) {
                        //合法则返回目标文件
                        return targetFile;
                    }
                    //如果快速定位失败，则从头到尾遍历文件队列
                    for (MappedFile tmpMappedFile : this.mappedFiles) {
                        //查找offset落在哪个内存映射文件中了
                        if (offset >= tmpMappedFile.getFileFromOffset()
                                && offset < tmpMappedFile.getFileFromOffset() + this.mappedFileSize) {
                            //返回找到的内存映射文件
                            return tmpMappedFile;
                        }
                    }
                }
                //如果遍历之后仍没找到对应的目标文件，那就根据条件判断是否可以返回队列首文件
                if (returnFirstOnNotFound) {
                    return firstMappedFile;
                }

            }
        } catch (Exception e) {
            log.error("findMappedFileByOffset Exception", e);
        }
        return null;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到文件队列中第一个内存映射文件的方法
     */
    public MappedFile getFirstMappedFile() {
        MappedFile mappedFileFirst = null;
        if (!this.mappedFiles.isEmpty()) {
            try {
                mappedFileFirst = this.mappedFiles.get(0);
            } catch (IndexOutOfBoundsException e) {

            } catch (Exception e) {
                log.error("getFirstMappedFile has exception.", e);
            }
        }
        return mappedFileFirst;
    }


    //根据偏移量查找内存映射文件的方法
    public MappedFile findMappedFileByOffset(final long offset) {
        return findMappedFileByOffset(offset, false);
    }


    //计算当前映射到内存中的数据的总大小
    public long getMappedMemorySize() {
        long size = 0;
        Object[] mfs = this.copyMappedFiles(0);
        if (mfs != null) {
            for (Object mf : mfs) {
                if (((ReferenceResource) mf).isAvailable()) {
                    size += this.mappedFileSize;
                }
            }
        }
        return size;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：重试删除文件队列首文件的方法
     */
    public boolean retryDeleteFirstFile(final long intervalForcibly) {
        MappedFile mappedFile = this.getFirstMappedFile();
        if (mappedFile != null) {
            if (!mappedFile.isAvailable()) {
                log.warn("the mappedFile was destroyed once, but still alive, " + mappedFile.getFileName());
                boolean result = mappedFile.destroy(intervalForcibly);
                if (result) {
                    log.info("the mappedFile re delete OK, " + mappedFile.getFileName());
                    List<MappedFile> tmpFiles = new ArrayList<>();
                    tmpFiles.add(mappedFile);
                    this.deleteExpiredFile(tmpFiles);
                } else {
                    log.warn("the mappedFile re delete failed, " + mappedFile.getFileName());
                }
                return result;
            }
        }
        return false;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：终止内存映射文件工作的方法
     */
    public void shutdown(final long intervalForcibly) {
        for (MappedFile mf : this.mappedFiles) {
            mf.shutdown(intervalForcibly);
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：销毁内存映射文件的方法
     */
    public void destroy() {
        for (MappedFile mf : this.mappedFiles) {
            mf.destroy(1000 * 3);
        }
        this.mappedFiles.clear();
        this.setFlushedWhere(0);
        File file = new File(storePath);
        if (file.isDirectory()) {
            file.delete();
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：执行内存交换的方法，该方法在第十四版本代码中没有被用到，我就先不添加注释了
     */
    @Override
    public void swapMap(int reserveNum, long forceSwapIntervalMs, long normalSwapIntervalMs) {
        if (mappedFiles.isEmpty()) {
            return;
        }
        if (reserveNum < 3) {
            reserveNum = 3;
        }
        Object[] mfs = this.copyMappedFiles(0);
        if (null == mfs) {
            return;
        }
        for (int i = mfs.length - reserveNum - 1; i >= 0; i--) {
            MappedFile mappedFile = (MappedFile) mfs[i];
            if (System.currentTimeMillis() - mappedFile.getRecentSwapMapTime() > forceSwapIntervalMs) {
                mappedFile.swapMap();
                continue;
            }
            if (System.currentTimeMillis() - mappedFile.getRecentSwapMapTime() > normalSwapIntervalMs
                    && mappedFile.getMappedByteBufferAccessCountSinceLastSwap() > 0) {
                mappedFile.swapMap();
            }
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：清理已交换的内存的方法
     */
    @Override
    public void cleanSwappedMap(long forceCleanSwapIntervalMs) {
        if (mappedFiles.isEmpty()) {
            return;
        }
        int reserveNum = 3;
        Object[] mfs = this.copyMappedFiles(0);
        if (null == mfs) {
            return;
        }
        for (int i = mfs.length - reserveNum - 1; i >= 0; i--) {
            MappedFile mappedFile = (MappedFile) mfs[i];
            if (System.currentTimeMillis() - mappedFile.getRecentSwapMapTime() > forceCleanSwapIntervalMs) {
                mappedFile.cleanSwapedMap(false);
            }
        }
    }

    //得到文件队列快照的方法
    public Object[] snapshot() {
        return this.mappedFiles.toArray();
    }


    //得到文件流的方法
    public Stream<MappedFile> stream() {
        return this.mappedFiles.stream();
    }

    //得到逆序文件流的方法
    public Stream<MappedFile> reversedStream() {
        return Lists.reverse(this.mappedFiles).stream();
    }


    public long getFlushedWhere() {
        return flushedWhere;
    }


    public void setFlushedWhere(long flushedWhere) {
        this.flushedWhere = flushedWhere;
    }


    public long getStoreTimestamp() {
        return storeTimestamp;
    }


    public void setStoreTimestamp(long storeTimestamp) {
        this.storeTimestamp = storeTimestamp;
    }


    public List<MappedFile> getMappedFiles() {
        return mappedFiles;
    }


    public int getMappedFileSize() {
        return mappedFileSize;
    }


    public long getCommittedWhere() {
        return committedWhere;
    }


    public void setCommittedWhere(final long committedWhere) {
        this.committedWhere = committedWhere;
    }


    public long getTotalFileSize() {
        return (long) mappedFileSize * mappedFiles.size();
    }


    public String getStorePath() {
        return storePath;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：根据偏移量范围获取相关内存映射文件列表的方法
     */
    public List<MappedFile> range(final long from, final long to) {
        Object[] mfs = copyMappedFiles(0);
        if (null == mfs) {
            return new ArrayList<>();
        }
        List<MappedFile> result = new ArrayList<>(mfs.length);
        for (Object mf : mfs) {
            MappedFile mappedFile = (MappedFile) mf;
            if (mappedFile.getFileFromOffset() + mappedFile.getFileSize() <= from) {
                continue;
            }
            if (to <= mappedFile.getFileFromOffset()) {
                break;
            }
            result.add(mappedFile);
        }
        return result;
    }
}
